package com.xujian.cdc.sink.starrocks;

import com.starrocks.connector.flink.StarRocksSink;
import com.starrocks.connector.flink.table.sink.StarRocksSinkOptions;
import com.ververica.cdc.connectors.mysql.source.MySqlSource;
import com.xujian.cdc.schema.JsonDebeziumDeserializationSchema;
import com.xujian.common.beans.JsonRecord;
import com.xujian.common.util.EnviromentUtil;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.api.java.utils.ParameterTool;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.streaming.api.functions.sink.SinkFunction;
import org.apache.flink.util.Collector;
import org.apache.flink.util.OutputTag;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.*;
import java.util.stream.Collectors;

/**
 * @author star xu
 * @date 2022/12/30 11:56
 * @Description: 解决日期格式和时差问题 TODO
 * @Slogan: 致敬大师, 致敬未来的自己
 */
public class Mysql2StarrocksCdc {

    private static final Logger log = LoggerFactory.getLogger(Mysql2StarrocksCdc.class);

    public static void main(String[] args) throws Exception {

        ParameterTool paramTool = ParameterTool.fromArgs(args);
       /* String tableName = paramTool.get("table");
        String srcHost = paramTool.get("srcHost");
        String srcDatabase = paramTool.get("srcDatabase");
        String srcUsername = paramTool.get("srcUsername");
        String srcPassword = paramTool.get("srcPassword");*/

        String srcHost = "127.0.0.1";
        String srcDatabase = "health100_source201607";
        String srcUsername = "tmp_select";
        String srcPassword = "qwer1234";
        String storgeType = "other";
        List<String> tableNames = new ArrayList<>();
//        tableNames.add("biz_diagnose_result_detail");
        tableNames.add("biz_reg_customer");
        String tableList = tableNames.stream().map(e -> srcDatabase + "." + e).collect(Collectors.joining(","));
        //1.创建执行环境
        StreamExecutionEnvironment env = EnviromentUtil.build(storgeType);

        //2.创建Sources数据源
        Properties properties = new Properties();
        properties.setProperty("converters", "dateConverters");
        properties.setProperty("dateConverters.type", "com.xujian.enter.converter.DateTimeConverter");

        MySqlSource<JsonRecord> mySqlSource = MySqlSource.<JsonRecord>builder()
                .hostname(srcHost)
                .port(3306)
                .databaseList(srcDatabase)
                .tableList(tableList)
                .username(srcUsername)
                .password(srcPassword)
                .serverTimeZone("Asia/Shanghai")
                //自定义序列化
                .deserializer(new JsonDebeziumDeserializationSchema())
                .debeziumProperties(properties)
                .build();

        //3.添加数据源
        DataStreamSource<JsonRecord> dataStreamSource = env.fromSource(mySqlSource, WatermarkStrategy.noWatermarks(), "Mysql source");

        //4.分流
        Map<String, OutputTag<String>> outputTagMap = new HashMap<>();
        for (String name : tableNames) {
            outputTagMap.put(name, new OutputTag<>(name, TypeInformation.of(String.class)));
        }

        SingleOutputStreamOperator<String> mainStream = dataStreamSource.process(new ProcessFunction<JsonRecord, String>() {
            @Override
            public void processElement(JsonRecord value, Context ctx, Collector<String> out) {
                int index = value.getTableName().lastIndexOf(".");
                String originalName = value.getTableName().substring(index + 1);
                ctx.output(outputTagMap.get(originalName), value.getFieldValue());
            }
        });
        //5.添加到不同的sink
        for (String name : tableNames) {
            DataStream<String> outputStream = mainStream.getSideOutput(outputTagMap.get(name));
            SinkFunction<String> sink = StarRocksSink.sink(
                    StarRocksSinkOptions.builder()
                            .withProperty("connector", "starrocks")
                            .withProperty("jdbc-url", "jdbc:mysql://192.168.16.77:9030?characterEncoding=utf-8&useSSL=false")
                            .withProperty("load-url", "192.168.16.77:8030")
                            .withProperty("username", "root")
                            .withProperty("password", "")
                            .withProperty("table-name", name + "_test")
                            .withProperty("database-name", "starrocks")
                            .withProperty("sink.buffer-flush.interval-ms", "10000")
                            .withProperty("sink.properties.format", "json")
                            .withProperty("sink.properties.strip_outer_array", "true")
                            .withProperty("sink.parallelism", "1")
                            .build()
            );
            outputStream.addSink(sink).name(name);
        }
        env.execute("监控mysql同步StarRocks表");
    }
}

